package com.chencong.online.driver.dwd;

import com.chencong.online.bean.ItemViewCountBean;
import com.chencong.online.bean.UserBehaviorBean;
import com.chencong.online.env.FlinkEnv;
import com.chencong.online.function.ItemCountAggFunc;
import com.chencong.online.function.TopNItemsCountProcessFunc;
import com.chencong.online.function.UserBehaviorBeanMapFunc;
import com.chencong.online.function.WindowItemCountResultFunc;
import com.chencong.online.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase;

import java.time.Duration;

/**
 * @Author chencong
 * @Description 热门统计
 * @Date 6:26 下午 2021/12/19
 * @Param
 **/
public class DwdHotItemsDriver {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = FlinkEnv.FlinkDataStreamRunEnv();
//        env.setStreamTimeCharacteristic(); 1.12后默认已经是事件事件语义
        //kafka配置
        String sourceTopic = "user_behavior_log";
        String groupId = "cc1";
        String offsetStrategy = "earliest";
        FlinkKafkaConsumerBase<String> flinkKafkaConsumer = KafkaUtil.getFlinkKafkaConsumer(sourceTopic, groupId, offsetStrategy);
        //todo 1、source
        DataStreamSource<String> userBehaviorInputDS = env.addSource(flinkKafkaConsumer);
//        userBehaviorInputDS.print("初始化数据");
        //todo 2、transform
        //转换为javabean，分配时间戳和watermark
        SingleOutputStreamOperator<UserBehaviorBean> userBehaviorBeanDS = userBehaviorInputDS.
                map(new UserBehaviorBeanMapFunc())
                .filter(data -> "pv".equals(data.getBehavior()))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<UserBehaviorBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserBehaviorBean>() {
                                    @Override
                                    public long extractTimestamp(UserBehaviorBean element, long recordTimestamp) {
                                        return element.getTimestamp() * 1000L;//watermark默认是200ms生成一次
                                    }
                                }));
//        userBehaviorBeanDS.print("转换为javabean后的数据");
        //分组开窗聚合，得到每个窗口内各个商品的count值
        SingleOutputStreamOperator<ItemViewCountBean> WindowItemIdAggCount = userBehaviorBeanDS
                .keyBy("itemId")    //品类分组
                .timeWindow(Time.hours(1), Time.minutes(5)) //开一时的窗口，5分钟滑动一次 注意窗口时[ )
                .aggregate(new ItemCountAggFunc(), new WindowItemCountResultFunc());
        //收集同一窗口的所有商品count数据，排序输出top n
        SingleOutputStreamOperator<String> resultStream = WindowItemIdAggCount
                .keyBy("windowEnd") //按照windowEnd分组
                .process(new TopNItemsCountProcessFunc(5));//排序后取前五
        resultStream.print("热度数据为");
        env.execute("DwdHotItems");
    }
}
